#include "NetIO.h"

Select_Reactor::Select_Reactor( void )
{
	TRACE_FUNC;
}

Select_Reactor::~Select_Reactor( void )
{
}

void Select_Reactor::cancel_handle( Socket handle )
{
	TRACE_FUNC;

	Handle_Data *data = handle.handle_data( );
	if( handle == INVALID_HANDLE || data == 0 )
		return;

	Operation_Queue op_queue;

	Guard<Synch_Mutex> guard_handle( data->mutex_ );
	for( int i = 0; i < OperationType::CONNECT; i ++ )
	{
		while( Operation *op = data->operations_[ i ].front( ) )
		{
			op->errcode( IOCANCELED );
			op_queue.push( op );
			data->operations_[ i ].pop( );
		}
	}
	data->is_cancel_ = true;
	guard_handle.release( );

	{
		Guard< Synch_Mutex > guard( mutex_ );
		complete_operations_.concat( &op_queue );

	}
	
	{
		Guard< Synch_Mutex > guard_map( handle_map_mutex_ );
		handle_map_.erase( handle );
	}

	interrupter_.interrupt( );
}

int Select_Reactor::register_handler_i( Socket &handle )
{
	return 0;
}

int Select_Reactor::start_op( Socket &handle, Operation *op )
{
	int op_type = op->optype( );
	if( op_type == OperationType::CONNECT )
		op_type = OperationType::WRITE;
	else if( op_type == OperationType::ACCEPT )
		op_type = OperationType::READ;

	Guard<Synch_Mutex> guard_handle( handle.handle_data( )->mutex_ );
	if ( handle.handle_data( )->is_cancel_ )
		return -1;

	handle.handle_data( )->operations_[ op_type ].push( op );
	interrupter_.interrupt( );
	return 0;
}

int Select_Reactor::collect_fdset( Handle_Set &read_set, Handle_Set &write_set, Handle_Set &except_set )
{
	Guard< Synch_Mutex > guard_map( handle_map_mutex_ );
	handle_map::iterator iter = handle_map_.begin( );
	for( ; iter != handle_map_.end( ); ++iter )
	{
		Guard< Synch_Mutex > guard_handle( iter->second->mutex_ );
		if( iter->second->is_cancel_ )
			continue;

		if( !iter->second->operations_[ OperationType::READ ].empty( ) )
			read_set.set_handle( iter->first );
		if( !iter->second->operations_[ OperationType::WRITE ].empty( ) )
		{
			write_set.set_handle( iter->first );
			if( iter->second->operations_[ OperationType::WRITE ].front( )->optype( ) == OperationType::CONNECT )
			{
				except_set.set_handle( iter->first );
			}
		}
	}
	return 0;
}


int Select_Reactor::run_once( )
{
	EventHandle maxfd;
	Handle_Set read_set,write_set,except_set;
	Operation_Queue op_queue;

	Guard<Synch_Mutex> guard( mutex_ );
	handle_all_complete_op( );
	guard.release( );

	if( exit_ )
		return -1;

	read_set.set_handle( interrupter_.read_handle( ) );
	collect_fdset( read_set, write_set, except_set );
	maxfd = read_set.maxfd( ) > write_set.maxfd( ) ? read_set.maxfd( ) :write_set.maxfd( );

	timeval timeout = {1, 0};
	int ret = ::select(( int ) maxfd+1, &read_set.get_fd_set( ),
	                   &write_set.get_fd_set( ), &except_set.get_fd_set( ), &timeout );

	if( ret > 0 )
	{
		if( FD_ISSET( interrupter_.read_handle( ), &read_set.get_fd_set( ) ) )
			interrupter_.reset( );

		Guard< Synch_Mutex > guard_map( handle_map_mutex_ );
		for( handle_map::iterator iter = handle_map_.begin( ); iter != handle_map_.end( ); ++iter )
		{
			Handle_Data *data = iter->second;
			Guard< Synch_Mutex > guard_handle( data->mutex_ );
			Operation *op = 0;
			if( FD_ISSET( iter->first, &read_set.get_fd_set( ) ) )
			{
				while( op = data->operations_[ OperationType::READ ].front( ) )
				{
					if( !op->perform( ) )
						break;
					data->operations_[ OperationType::READ ].pop( );
					op_queue.push( op );
				}
			}

			if( FD_ISSET( iter->first, &write_set.get_fd_set( ) ) ||
			        FD_ISSET( iter->first, &except_set.get_fd_set( ) ) )
			{
				while( op = data->operations_[ OperationType::WRITE ].front( ) )
				{
					if( !op->perform( ) )
						break;
		
					data->operations_[ OperationType::WRITE ].pop( );
					op_queue.push( op );
				}
			}
		}
		guard_map.release( );
	}

	guard.acquire( );
	complete_operations_.concat( &op_queue );
	handle_all_complete_op( );
	return 0;
}
